package index;

import global.AttrType;
import global.Convert;
import global.IndexType;
import global.RID;
import global.Size;
import global.TID;
import heap.FileAlreadyDeletedException;
import heap.HFBufMgrException;
import heap.HFDiskMgrException;
import heap.HFException;
import heap.Heapfile;
import heap.InvalidSlotNumberException;
import heap.InvalidTupleSizeException;
import heap.Scan;
import heap.Tuple;
import iterator.CondExpr;
import iterator.Iterator;
import iterator.Projection;
import iterator.UnknownKeyTypeException;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;

import bitmap.BitMapFile;
import bitmap.InstantiateColumnarFile;
import bitmap.KeyClass;
import btree.BT;
import btree.BTFileScan;
import btree.BTreeFile;
import btree.IndexFile;
import btree.IndexFileScan;
import btree.IntegerKey;
import btree.KeyDataEntry;
import btree.LeafData;
import btree.StringKey;
import columnar.Columnarfile;
import columnar.TupleScan;
import diskmgr.pcounter;

public class ColumnarIndexScan  extends Iterator{
	
	private AttrType type;
	private IndexType index;
	private IndexFile     indFile;
	private IndexFileScan indScan;
	private BitMapFile bmf=null;
	private BTreeFile btf=null;
	private Scan bitMapScan=null;
	private ArrayList  projections;
	private Columnarfile c = null;
	
	private static int startRead = 0, startWrite = 0;

//	public FldSpec[]      perm_mat;
	private AttrType    _type;
	private short       _s_size=0; 
//	private CondExpr[]    _selects;
	private int           _noInFlds;
//	private int           _noOutFlds;
	private Heapfile      f;
	private Tuple         tuple1;
	private Tuple         Jtuple;
//	private int           t1_size;
	private boolean       index_only;    
	private CondExpr[] selects;
	private String relName;
	public String getRelName() {
		return relName;
	}

	public String getIndName() {
		return indName;
	}

	private String indName;
	public CondExpr[] getSelects() {
		return selects;
	}

	public void setSelects(CondExpr[] selects) {
		this.selects = selects;
	}

	public AttrType getType() {
		return type;
	}

	public void setType(AttrType type) {
		this.type = type;
	}

	public IndexType getIndex() {
		return index;
	}

	public void setIndex(IndexType index) {
		this.index = index;
	}

	 public static int getChoice () {
		    
		    BufferedReader in = new BufferedReader (new InputStreamReader(System.in));
		    int choice = -1;
			
		    try {
		      choice = Integer.parseInt(in.readLine());
		    }
		    catch (NumberFormatException e) {
		      return -1;
		    }
		    catch (IOException e) {
		      return -1;
		    }
		    
		    return choice;
		  } 
	
	public ColumnarIndexScan(IndexType index,String relName,String indName,	AttrType type,	short str_sizes,CondExpr[] selects,
			boolean indexOnly) throws IndexException, UnknownIndexTypeException
	{
	    index_only = indexOnly;  
	    _type = type;
	    _s_size= str_sizes;
	    _noInFlds=1;
	    try {
			f=new Heapfile(relName);
			
		} catch (Exception e) {
			e.printStackTrace();
		}

		switch(index.indexType) {

		case IndexType.B_Index:
			try {
				btf = new BTreeFile(indName); 
				indScan = (BTFileScan) IndexUtils.BTree_scan(selects, btf);
			}
			catch (Exception e) {
				throw new IndexException(e, "IndexScan.java: BTreeFile exceptions caught from IndexUtils.BTree_scan().");
			}
			break;

		case IndexType.Bitmap:
			try {
				bmf=new BitMapFile(indName);
				bitMapScan=bmf.openScan();
			} catch (Exception e) {
				throw new IndexException(e, "BitMap creation Error!!");
			}	    	
			break;

		case IndexType.None:
			break;

		default:
			throw new UnknownIndexTypeException("Only BTree and Bitmap indexes are supported so far");

		}

		this.setIndex(index);
		this.setSelects(selects);
		this.setType(type);
	}
	public ColumnarIndexScan(
	String relName,
	int [] fldNum,
	IndexType [] index,
	String [] indName,
	AttrType [] types,
	short [] str_sizes,
	String [] columnName,
	btree.KeyClass [] value,
	String [] operator,
	boolean indexOnly,
	Columnarfile c,
	String [] conditions,
	ArrayList projections
	)
	{
		BTreeFile [] btf = new BTreeFile[4];
		BitMapFile [] bmf = new BitMapFile[4];
		KeyDataEntry kde;
		int iBtfCount = 0;
		int iBmfCount = 0;
		int keyType=0;
		btree.KeyClass lowkey=null;
		btree.KeyClass hikey=null;
		startRead = pcounter.rcounter;
		startWrite = pcounter.wcounter;
		BitSet [] bs = new BitSet[index.length];
		KeyDataEntry entry;
		BTFileScan btscan = null;
		this.projections = projections;
		this.c = c;
		try {
			if (indexOnly)
			{
				for (int i = 0 ; i < 2 ; i++)
				{
					
					Heapfile colFile =c.getHeapfileForColumname(columnName[i]);
					if (index[i].indexType == IndexType.B_Index)
					{
						try 
						{
							Tuple hTuple=null;
								
							if(value[i] instanceof btree.IntegerKey)
							{
					
								btf[iBtfCount] = new BTreeFile("btree"+columnName[i], AttrType.attrInteger, 4, 1/*delete*/);
								
								keyType=AttrType.attrInteger;
							}
							if(value[i] instanceof btree.StringKey)
							{
					
								btf[iBtfCount] = new BTreeFile("btree"+columnName[i], AttrType.attrString, 60, 1/*delete*/);
								keyType=AttrType.attrString;
							}
							
							bs[iBtfCount] = new BitSet();

							//BT.printBTree(btf[iBtfCount].getHeaderPage());
							System.out.println("printing leaf pages...");

							  
							btree.KeyClass tempKey;
							
							System.out.println("Please input the LOWER key : ");
							  tempKey=(btree.KeyClass)Helper.getChoice();
							  if(tempKey instanceof IntegerKey)
							  {
								  
								  lowkey=(btree.IntegerKey)tempKey;
							  }
							  else if(tempKey instanceof btree.StringKey)
							  {
								  
								  lowkey=(btree.StringKey)tempKey;
							  }
							  
							  System.out.println("Please input the HIGHER key : ");
							  tempKey=(btree.KeyClass)Helper.getChoice();
							  if(tempKey instanceof IntegerKey)
							  {
								
								  hikey=(btree.IntegerKey)tempKey;
							  }
							  else if(tempKey instanceof btree.StringKey)
							  {
								
								  hikey=(btree.StringKey)tempKey;
							  }
								  
							  
							  btscan = btf[iBtfCount].new_scan(lowkey, hikey);
							
							while((entry = btscan.get_next())!=null)
							  {
								  if(entry!=null)
								  {
									  char sw=operator[i].charAt(0);
									  
									  boolean eval=false;
									  switch(sw)
									  {
									  case '>':
										  if(entry.key instanceof IntegerKey)
										  {
											  if(((IntegerKey)entry.key).getKey()>((btree.IntegerKey)value[i]).getKey())
												  eval=true;
											  else
												  eval=false;
										  }
										  else if(entry.key instanceof StringKey)
										  {
											  if(((StringKey)entry.key).getKey().compareTo(((btree.StringKey)value[i]).getKey())>0)
												  eval=true;
											  else
												  eval=false;
										  }
										  break;
									  case '<':
										  if(entry.key instanceof IntegerKey)
										  {
											  if(((IntegerKey)entry.key).getKey()<((btree.IntegerKey)value[i]).getKey())
												  eval=true;
											  else
												  eval=false;
										  }
										  else if(entry.key instanceof StringKey)
										  {
											  if(((StringKey)entry.key).getKey().compareTo(((btree.StringKey)value[i]).getKey())>0)
												  eval=true;
											  else
												  eval=false;
										  }
										  break;
									  case '!':
										  if(entry.key instanceof IntegerKey)
										  {
											  if(((IntegerKey)entry.key).getKey()>((btree.IntegerKey)value[i]).getKey())
												  eval=true;
											  else
												  eval=false;
										  }
										  else if(entry.key instanceof StringKey)
										  {
											  if(((StringKey)entry.key).getKey().compareTo(((btree.StringKey)value[i]).getKey())!=0)
												  eval=true;
											  else
												  eval=false;
										  }
										  break;
									  case '=':
										  if(entry.key instanceof IntegerKey)
										  {
											  if(((btree.IntegerKey)entry.key).getKey().equals(((btree.IntegerKey)value[i]).getKey()))
											  {
												  bs[iBtfCount].set(colFile.RidToPos(((btree.LeafData)entry.data).getData()));
												 //System.out.println("data at position"+Convert.getIntValue(0,colFile.getRecord(((btree.LeafData)entry.data).getData()).getData()));
											  }
										  }
										  else if(entry.key instanceof StringKey)
										  {
											  System.out.println("btree scan"+((StringKey)entry.key).getKey());
											  System.out.println("value "+(((btree.StringKey)value[i]).getKey()));
											  if(((StringKey)entry.key).getKey().compareTo(((btree.StringKey)value[i]).getKey())==0)
											  {
												  int pos = colFile.RidToPos(((btree.LeafData)entry.data).getData());
												  bs[iBtfCount].set(pos);
												  getProjection(pos);
												  //System.out.println("data at position "+pos+" "+Convert.getStrValue(0,colFile.getRecord(((btree.LeafData)entry.data).getData()).getData(),Size.STRINGSIZE));
											  }
											  
										  }
										  break;

									  }
								  }
							  }
							  	
						} catch (Exception e) 
						{
							// TODO Auto-generated catch block
							e.printStackTrace();
						}
						
						System.out.println("bitset of btree"+bs[iBtfCount]);
						iBtfCount++;
					
					}
					else if (index[i].indexType == IndexType.Bitmap)
					{
						Tuple bTuple=new Tuple();
						RID bRID=new RID();
						byte[] yes=new byte[2];
						byte[] no=new byte[2];
						Convert.setCharValue('0', 0, no);
						Convert.setCharValue('1', 0, yes);
						
						int columnIndex = 1;
						KeyClass value1 = null;
						String bmfName = indName[i];
						
						bmf[iBtfCount] = InstantiateColumnarFile.getColumnarFile(null, c, value1, columnIndex,bmfName, colFile);
						Scan bScan = bmf[iBtfCount].openScan();
						int position=1;
						bs[iBtfCount] = new BitSet();
						while((bTuple=bScan.getNext(bRID))!=null)
						{
							if(Arrays.equals(bTuple.getData(),yes)==true)
							{
								bs[iBtfCount].set(position);
								
							}
							position++;
						}
						System.out.println("bitset of bitmap"+bs[iBtfCount]);
						iBtfCount++;
					}
					
				}
				
				for (int i = 0 ; i < conditions.length; i++)
				{
					if(conditions[i]!=null)
					{
					//System.out.println("index"+i);
					if (conditions[i].equalsIgnoreCase("And"))
					{
						bs[0].and(bs[i+1]);
						System.out.println("And Result:");
						System.out.println(bs[0]);
					}
					else if (conditions[i].equalsIgnoreCase("or"))
					{
						bs[0].or(bs[i+1]);
						System.out.println("Or Result:");
						System.out.println(bs[0]);
					}
					else if (conditions[i].equalsIgnoreCase("xor"))
					{
						bs[0].xor(bs[i+1]);
						System.out.println("Xor Result:");
						System.out.println(bs[0]);
					}
					}
				}
				System.out.println("Final Result:");
				for (int i = bs[0].nextSetBit(0) ; i > 0; i= bs[0].nextSetBit(i+1))
				{
					//System.out.println("position"+i);
					getProjection(i);
				}
				
				
				
			}
			else
			{
				TupleScan t = new TupleScan(c);
				Tuple t1 = new Tuple();
				TID tid = new TID();
				tid.recordIDs = new RID[c.getnumColumns()];
				for (int i = 0 ; i < c.getnumColumns(); i++)
				{
					tid.recordIDs[i] = new RID();
			
				}
				boolean [] bs1 = new boolean[conditions.length];
				for (int i = 0 ; i <conditions.length ; i++)
				{
					bs1[i] = false;
				}
				boolean result = false;
				int k = 0;
				while((t1 = t.getNext(tid))!=null)
				{
					t1.setTupleMetaData(c.getTupleLength(), (short) c.getnumColumns(), c.getFldOffset());
					for (int i = 0 ; i < 2; i++)
					{
						if (fldNum[i]!=0)
						{
							if(value[i] instanceof btree.IntegerKey)
							{
								if(t1.getIntFld(fldNum[i])==(((btree.IntegerKey)value[i]).getKey()))
								{
									bs1[i] = true;
								}
							}
							if(value[i] instanceof btree.StringKey)
							{
								if(t1.getStrFld(fldNum[i]).equalsIgnoreCase(((btree.StringKey)value[i]).getKey()))
								{
									bs1[i] = true;
									
								}
								
							}
							
						}
					}
					result = bs1[0]; 
					//System.out.println("result "+result);
					for (int i = 0 ; i < conditions.length; i++)
					{
						if(conditions[i]!=null)
						{
							
						if (conditions[i].equalsIgnoreCase("And"))
						{
							result = result && bs1[i+1];							
						}
						else if (conditions[i].equalsIgnoreCase("or"))
						{
							result = result || bs1[i+1];
						}
						else if (conditions[i].equalsIgnoreCase("xor"))
						{
							result = result | bs1[i+1];
						}
						}
					}
					
					if (result)
					t1.print(types);
					
					k++;
					for (int i = 0 ; i <conditions.length ; i++)
					{
						bs1[i] = false;
					}
					result = false;
					
				}
				
				
			}
			
		
		
		
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}catch (Exception e) {
			// TODO Auto-generated catch block
			System.out.println("in e");
			e.printStackTrace();
		}
		System.out.println("Disk Reads"+ (pcounter.rcounter - startRead));
		System.out.println("Disk Writes"+ (pcounter.wcounter - startWrite));
		
	}
	
	public  TID getProjection(int position) throws InvalidSlotNumberException, HFException, HFDiskMgrException, HFBufMgrException, Exception
	{
		int colSize  = projections.size();
		ArrayList<Tuple> arrTuples=new ArrayList<Tuple>();
		RID rid = new RID();
		Heapfile hf;
		Tuple t;
		TID tid=new TID(colSize);
		tid.recordIDs=new RID[colSize];
		tid.pos=position;
		tid.numRIDs=colSize;
		System.out.print("[");
		for(int j=0;j<colSize;j++)
			tid.recordIDs[j]=new RID();

		System.out.print("[");
		for (int i = 0 ; i < colSize ; i++)
		{
			hf = c.getHeapfileForColumname(projections.get(i).toString());
			rid = hf.PosToRid(position);
			t = hf.getRecord(rid);
			tid.recordIDs[i]=rid;
			if(t.getLength()>4)
				System.out.print(Convert.getStrValue(0, t.getData(), t.getLength())+", ");
			else
				System.out.print(Convert.getIntValue(0, t.getData())+", ");
			arrTuples.add(hf.getRecord(rid));
		}
		System.out.println("]");
		return tid;
	}
	  public Tuple get_next() 
	    throws IndexException, 
		   UnknownKeyTypeException,
		   IOException
	  {
	    RID rid=null;
	    KeyDataEntry nextentry = null;
	    
	    if(btf!=null)
	    {
	    try {
	      nextentry = indScan.get_next();
	    }
	    catch (Exception e) {
	      throw new IndexException(e, "IndexScan.java: BTree error");
	    }	  
	    
	      if (index_only) {
		// only need to return the key 

		AttrType[] attrType = new AttrType[1];
		short[] s_sizes = new short[1];
		
		if (_type.attrType == AttrType.attrInteger) {
		  attrType[0] = new AttrType(AttrType.attrInteger);
		  try {
		    Jtuple.setHdr((short) 1, attrType, s_sizes);
		  }
		  catch (Exception e) {
		    throw new IndexException(e, "IndexScan.java: Heapfile error");
		  }
		  
		  try {
		    Jtuple.setIntFld(1, ((IntegerKey)nextentry.key).getKey().intValue());
		  }
		  catch (Exception e) {
		    throw new IndexException(e, "IndexScan.java: Heapfile error");
		  }	  
		}
		else if (_type.attrType == AttrType.attrString) {
		  
		  attrType[0] = new AttrType(AttrType.attrString);
		  // calculate string size of _fldNum
		  int count = 0;
		  for (int i=0; i<1; i++) {
		    if (_type.attrType == AttrType.attrString)
		      count ++;
		  } 
		  s_sizes[0] = _s_size;
		  
		  try {
		    Jtuple.setHdr((short) 1, attrType, s_sizes);
		  }
		  catch (Exception e) {
		    throw new IndexException(e, "IndexScan.java: Heapfile error");
		  }
		  
		  try {
		    Jtuple.setStrFld(1, ((StringKey)nextentry.key).getKey());
		  }
		  catch (Exception e) {
		    throw new IndexException(e, "IndexScan.java: Heapfile error");
		  }	  
		}
		else {
		  // attrReal not supported for now
		  throw new UnknownKeyTypeException("Only Integer and String keys are supported so far"); 
		}
		return Jtuple;
	      }
	      
	      rid = ((LeafData)nextentry.data).getData();
	      try {
	    	  tuple1 = f.getRecord(rid);
	      }
	      catch (Exception e) {
	    	  throw new IndexException(e, "IndexScan.java: getRecord failed");
	      }
	      
	      
	      
	      return tuple1;
	    
	    }
	    else if(bmf!=null)
	    {
	    	try {
				Tuple t=bitMapScan.getNext(rid);
				return t;
			} catch (Exception e) {
				e.printStackTrace();
			}
	    }
	    
	    return null; 
	  }

	  public void close() throws IOException, IndexException
	  {
	    if (!closeFlag) {
	      if (indScan instanceof BTFileScan) {
		try {
		  ((BTFileScan)indScan).DestroyBTreeFileScan();
		}
		catch(Exception e) {
		  throw new IndexException(e, "BTree error in destroying index scan.");
		}
	      }
			else if(bmf!=null)
			{
				try {
					bmf.deleteFile();
				} catch (InvalidSlotNumberException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (FileAlreadyDeletedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (InvalidTupleSizeException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (HFBufMgrException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (HFDiskMgrException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}

	      
	      closeFlag = true; 
	    }
	  }
}
